extern crate core;
#[macro_use]
extern crate lazy_static;
extern crate log;
extern crate paste;
extern crate rbatis;
extern crate rbs;
extern crate time;
extern crate walkdir;
use mio::net::{TcpListener, TcpStream};
use std::cell::RefCell;
use std::cmp::Ordering::Equal;
use std::collections::vec_deque::VecDeque;
use std::collections::HashMap;
use std::error::Error;
use std::fmt::{Debug, Display, Formatter};
use std::fs::{File, OpenOptions};
use std::io::{stdin, stdout, BufRead, BufReader, BufWriter, Read, Write};
use std::marker::PhantomData;
use std::mem::take;
use std::net::SocketAddr::V4;
use std::net::{Shutdown, SocketAddrV4};
use std::ops::{Deref, DerefMut};
use std::process::Termination;
use std::ptr::null;
use std::rc::Rc;
use std::str::{from_utf8, FromStr};
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{mpsc, Arc, Mutex};
use std::thread;
use std::thread::available_parallelism;
use std::time::SystemTime;

use actix_web::body::{BoxBody, MessageBody};
use actix_web::http::header::ContentType;
use actix_web::web::{Json, Path};
use actix_web::{
    get, middleware, post, web, App, HttpRequest, HttpResponse, HttpServer, Responder,
};
use anyhow::{anyhow, Result};
use ftlog::appender::{FileAppender, Period};
use ftlog::FtLogFormatter;
use log::{debug, error, info, log, trace, warn, Log};
use mio::event::Event;
use mio::{Events, Interest, Poll, Registry, Token};
use rand::Rng;
use rbatis::dark_std::err;
use rbatis::executor::RbatisRef;
use rbatis::html_sql;
use rbatis::rbdc::decimal::Decimal;
use rbatis::sql::{Page, PageRequest};
use rbatis::{
    crud, impl_delete, impl_insert, impl_select, impl_select_page, impl_update, impled, Rbatis,
};
use rbdc_mysql::driver::MysqlDriver;
use rbdc_mysql::options::MySqlConnectOptions;
use rbdc_sqlite::driver::SqliteDriver;
use regex::{Captures, Regex};
use rusqlite::{params, Connection};
use serde::{Deserialize, Serialize};
use serde_json::json;
use time::Duration;
use walkdir::DirEntry;

use crate::config::{ftlog_init, load_from_yaml, RootAppConfig, SqliteConf};
use crate::extern_lib::*;
use crate::model::TsAreas;
use crate::utils::macros_func::get_all_trait;
use crate::utils::singleton::SingleInit;

mod alg;
mod config;
mod extern_lib;
mod model;
mod service;
mod test_only;
pub mod utils;

const CONFIG_FILE: &str = "src/app.yml";

lazy_static! {
    pub static ref ROOT_APP_CONFIG: config::RootAppConfig = RootAppConfig::new("src/app.yml");
    pub static ref RB: rbatis::Rbatis = {
        let mut rb = Rbatis::new();
        if let Some(SqliteConf { url }) = &ROOT_APP_CONFIG.sqlite {
            let conf = ROOT_APP_CONFIG.mysql.as_ref().expect("miss mysql config");
            rb.init_opt(
                MysqlDriver {},
                MySqlConnectOptions::new()
                    .host(conf.host.as_deref().expect("miss host"))
                    .port(conf.port.expect("miss port"))
                    .password(conf.password.as_deref().expect("miss password"))
                    .username(conf.username.as_deref().expect("miss username"))
                    .database(conf.database.as_deref().expect("miss database")),
            )
            .unwrap();
        }
        rb
    };
}
#[post("/hello")]
async fn hello() -> String {
    rand::random::<i32>().to_string()
}
#[actix_web::main]
async fn main() -> std::io::Result<()> {
    ftlog_init(&ROOT_APP_CONFIG).unwrap();
    let config::ActixConf { server, port } = ROOT_APP_CONFIG.actix.as_ref().unwrap();
    info!("{}", "server is started now");
    HttpServer::new(|| {
        App::new()
            // .wrap(middleware::Logger::default())
            // .service(service::query_first)
            // .service(service::query_trait)
            // .service(service::query_trait_code)
            .service(hello)
    })
    .workers(available_parallelism().unwrap().get())
    .bind((server.as_deref().unwrap(), port.unwrap()))
    .expect("bind server port failed!")
    .run()
    .await
}
const SERVER: Token = Token(0);

// Some data we'll send over the connection.
const DATA: &[u8] = b"Hello world!\n";

// #[tokio::main]
// pub async fn main() {
//     ftlog_init(&ROOT_APP_CONFIG).unwrap();
//     let vec = service::select_by_condition(&mut RB.clone(), "110101002", "110101002".len() as i32).await;
//     info!("result colume is {:?}", vec);
// }

// pub fn main() -> std::io::Result<()> {
//     ftlog_init(&ROOT_APP_CONFIG).unwrap();
//     // Create a poll instance.
//     let mut poll = Poll::new()?;
//     // Create storage for events.
//     let mut events = Events::with_capacity(128);
//
//     // Setup the TCP server socket.
//     let addr = "127.0.0.1:9000".parse().unwrap();
//     let mut server = mio::net::TcpListener::bind(addr)?;
//
//     // Register the server with poll we can receive events for it.
//     poll.registry()
//         .register(&mut server, SERVER, Interest::READABLE)?;
//
//     // Map of `Token` -> `TcpStream`.
//     let mut connections = HashMap::new();
//     // Unique token for each incoming connection.
//     let mut unique_token = Token(SERVER.0 + 1);
//
//     println!("You can connect to the server using `nc`:");
//     println!(" $ nc 127.0.0.1 9000");
//     println!("You'll see our welcome message and anything you type will be printed here.");
//
//     loop {
//         poll.poll(&mut events, None)?;
//
//         for event in events.iter() {
//             match event.token() {
//                 SERVER => loop {
//                     // Received an event for the TCP server socket, which
//                     // indicates we can accept an connection.
//                     let (mut connection, address) = match server.accept() {
//                         Ok((connection, address)) => (connection, address),
//                         Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
//                             // If we get a `WouldBlock` error we know our
//                             // listener has no more incoming connections queued,
//                             // so we can return to polling and wait for some
//                             // more.
//                             break;
//                         }
//                         Err(e) => {
//                             // If it was any other kind of error, something went
//                             // wrong and we terminate with an error.
//                             return Err(e);
//                         }
//                     };
//
//                     println!("Accepted connection from: {}", address);
//
//                     let token = next(&mut unique_token);
//                     poll.registry().register(
//                         &mut connection,
//                         token,
//                         Interest::READABLE.add(Interest::WRITABLE),
//                     )?;
//
//                     connections.insert(token, connection);
//                 },
//                 token => {
//                     // Maybe received an event for a TCP connection.
//                     let done = if let Some(connection) = connections.get_mut(&token) {
//                         handle_connection_event(poll.registry(), connection, event)?
//                     } else {
//                         // Sporadic events happen, we can safely ignore them.
//                         false
//                     };
//                     if done {
//                         if let Some(mut connection) = connections.remove(&token) {
//                             poll.registry().deregister(&mut connection)?;
//                         }
//                     }
//                 }
//             }
//         }
//     }
// }

fn next(current: &mut Token) -> Token {
    let next = current.0;
    current.0 += 1;
    Token(next)
}

/// Returns `true` if the connection is done.
fn handle_connection_event(
    registry: &Registry,
    connection: &mut TcpStream,
    event: &Event,
) -> std::io::Result<bool> {
    if event.is_writable() {
        // We can (maybe) write to the connection.
        match connection.write(DATA) {
            // We want to write the entire `DATA` buffer in a single go. If we
            // write less we'll return a short write error (same as
            // `io::Write::write_all` does).
            Ok(n) if n < DATA.len() => return Err(std::io::ErrorKind::WriteZero.into()),
            Ok(_) => {
                // After we've written something we'll reregister the connection
                // to only respond to readable events.
                registry.reregister(connection, event.token(), Interest::READABLE)?
            }
            // Would block "errors" are the OS's way of saying that the
            // connection is not actually ready to perform this I/O operation.
            Err(ref err) if would_block(err) => {}
            // Got interrupted (how rude!), we'll try again.
            Err(ref err) if interrupted(err) => {
                return handle_connection_event(registry, connection, event);
            }
            // Other errors we'll consider fatal.
            Err(err) => return Err(err),
        }
    }

    if event.is_readable() {
        let mut connection_closed = false;
        let mut received_data = vec![0; 4096];
        let mut bytes_read = 0;
        // We can (maybe) read from the connection.
        loop {
            match connection.read(&mut received_data[bytes_read..]) {
                Ok(0) => {
                    // Reading 0 bytes means the other side has closed the
                    // connection or is done writing, then so are we.
                    connection_closed = true;
                    break;
                }
                Ok(n) => {
                    bytes_read += n;
                    if bytes_read == received_data.len() {
                        received_data.resize(received_data.len() + 1024, 0);
                    }
                }
                // Would block "errors" are the OS's way of saying that the
                // connection is not actually ready to perform this I/O operation.
                Err(ref err) if would_block(err) => break,
                Err(ref err) if interrupted(err) => continue,
                // Other errors we'll consider fatal.
                Err(err) => return Err(err),
            }
        }

        if bytes_read != 0 {
            let received_data = &received_data[..bytes_read];
            if let Ok(str_buf) = from_utf8(received_data) {
                println!("Received data: {}", str_buf.trim_end());
            } else {
                println!("Received (none UTF-8) data: {:?}", received_data);
            }
        }

        if connection_closed {
            println!("Connection closed");
            return Ok(true);
        }
    }

    Ok(false)
}

fn would_block(err: &std::io::Error) -> bool {
    err.kind() == std::io::ErrorKind::WouldBlock
}

fn interrupted(err: &std::io::Error) -> bool {
    err.kind() == std::io::ErrorKind::Interrupted
}

// fn main() {
//     ftlog_init(&ROOT_APP_CONFIG).unwrap();
//     time_cost(|| {
//         for x in 0..10000 {
//             info!("loop test");
//         }
//     });
// }

fn do_handle_request_stream(mut p0: TcpStream) {
    let mut buffer = [0 as u8; 1024];
    p0.read(&mut buffer).unwrap();
    log_info!("request info :\n{}", String::from_utf8_lossy(&buffer[..]));

    let contents = r#"<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="utf-8">
    <title>Hello!</title>
  </head>
  <body>
    <h1>Hello!</h1>
    <p>Hi from Rust</p>
  </body>
</html>"#;

    let response = format!(
        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{}",
        contents.len(),
        contents
    );

    p0.write(response.as_bytes()).unwrap();
    p0.flush().unwrap();
}

#[cfg(test)]
mod main_test {
    use std::borrow::BorrowMut;
    use std::fs::create_dir_all;
    use std::io::Write;
    use std::mem::replace;
    use std::path::{Path, PathBuf};
    use std::sync::{Arc, Condvar, Mutex, RwLock, TryLockResult};
    use std::time::Duration;
    use std::{env, println};

    use walkdir::WalkDir;

    use crate::utils::macros_func::has_traits;

    use super::*;

    // fn create_new_file_append(dir: &str, file: &str) -> File {}

    #[test]
    fn test_shit() {
        submit_all!(
            {
                thread::sleep(Duration::from_secs(4));
                println!("first");
            },
            {
                thread::sleep(Duration::from_secs(4));
                println!("second");
            }
        );

        let x = commit!({
            let mut unsort_vec = vec![2, 4, 6, 8, 10, 9, 7, 5, 3, 1];
            unsort_vec.sort();
            BufReader::new(File::open("src/repos/traits_index.txt").unwrap())
        });

        std::println!("wait for sort");
        std::println!("{:?}", x.get());
        flush_thread_pool!();
    }

    #[test]
    fn it_works() {
        let pool = SimplePool::new(4);
        pool.execute(|| {
            for _ in 1..=100 {
                DBG!("now ");
            }
        });
    }

    #[test]
    fn test2() {
        let pool = SimplePool::new(4);
        let rwlock = Arc::new(RwLock::new(0 as usize));
        for _ in 0..4 {
            let crw = rwlock.clone();
            pool.execute(move || {
                for _ in 0..10 {
                    if let Ok(x) = crw.try_read() {
                        println!("{}", x);
                    } else {
                        thread::sleep(Duration::from_millis(1));
                    }
                }
            });
        }

        pool.execute(move || {
            if let Ok(mut x) = rwlock.try_write() {
                *x += 1;
            }
        });
    }

    #[test]
    fn test3() {
        let pool = SimplePool::new(4);
        // let atomic_bool = Arc::new(Mutex::new(false));
        // let cond = Arc::new(Condvar::new());

        for _ in 0..4 {
            // let ccond = cond.clone();
            // let catomic = atomic_bool.clone();
            pool.execute(move || {
                loop {
                    // let mut guard = ccond.wait_while(catomic.lock().unwrap(), |g| {
                    //     false
                    // }).unwrap();
                    info!("{:?}print now!", thread::current().id());
                    // *guard = false;
                }
            });
        }

        thread::sleep(Duration::from_secs(4));
        // println!("{:?},print now!", thread::current().id());

        // for _ in 0..3 {
        //     thread::sleep(Duration::from_secs(1));
        //     *atomic_bool.lock().unwrap() = true;
        //     cond.notify_one();
        // }
    }

    #[test]
    pub fn find_json() {
        let source_dir = r#"D:\CPP_TOOL\Rust\cargo\registry\src"#;

        WalkDir::new(source_dir)
            .into_iter()
            .filter_entry(|entry| {
                (entry.file_type().is_dir()
                    && !(entry.file_name().eq_ignore_ascii_case("test")
                        || entry.file_name().eq_ignore_ascii_case("benches")))
                    || (entry.file_type().is_file()
                        && entry.file_name().to_str().unwrap().ends_with(".rs"))
            })
            .for_each(|e| {
                if e.as_ref().unwrap().file_type().is_file() {
                    let entr = e.unwrap();
                    BufReader::new(File::open(entr.clone().path()).unwrap())
                        .lines()
                        .into_iter()
                        .for_each(|strs| {
                            if strs.unwrap().contains("json!") {
                                println!("{}", entr.path().as_os_str().to_str().unwrap());
                            }
                        })
                }
            });
    }
}

#[test]
pub fn test_mio_client() {
    ftlog_init(&ROOT_APP_CONFIG).unwrap();
    use core::time::Duration;
    loop {
        let mut stream = std::net::TcpStream::connect_timeout(
            &V4(SocketAddrV4::from_str("127.0.0.1:9000").unwrap()),
            Duration::from_secs(100),
        )
        .unwrap();
        let mut buf = [0 as u8; 1024];
        for _ in 0..1 {
            stream.write_all("hello".as_bytes()).unwrap();
        }

        stream.read(&mut buf).unwrap();
        info!("{}", String::from_utf8_lossy(&mut buf));

        stream.shutdown(Shutdown::Both).unwrap();
        std::thread::sleep(Duration::from_secs(2));
    }
}
